package jmine.tec.datadigester.pacoteimportacao.processor;

import java.io.InputStream;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

import jmine.tec.datadigester.DataDigesterMessages;
import jmine.tec.datadigester.pacoteimportacao.beans.PacoteImportacaoBean;
import jmine.tec.datadigester.pacoteimportacao.beans.PacoteImportacaoFile;
import jmine.tec.executionlog.ExecutionLogController;

import org.apache.commons.io.IOUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import bancosys.tec.datadigester.DataDigesterController;
import bancosys.tec.datadigester.DataDigesterException;
import bancosys.tec.datadigester.DigesterBean;
import bancosys.tec.datadigester.processor.AbstractProcessor;
import bancosys.tec.rtm.impl.RtmController;

/**
 * Processor que trata o bean PacoteImportacaoBean.
 * 
 * @author Rafael Volpato
 */
public class PacoteImportacaoProcessor extends AbstractProcessor {

    private static final Log LOG = LogFactory.getLog(PacoteImportacaoProcessor.class);

    private DataDigesterController dataDigesterController;

    private ExecutionLogController executionLogController;

    private RtmController rtmController;

    /**
     * Executa o processamento do PacoteImportacaoBean.
     * 
     * @param bean PacoteImportacaoBean
     */
    @Override
    protected void doProcess(DigesterBean bean) {
        PacoteImportacaoBean pacoteIntegracaoBean = (PacoteImportacaoBean) bean;
        if (this.executionLogController.isExecuting()) {
            this.executionLogController.setTotalItemsInExecution(pacoteIntegracaoBean.getNumberOfItems());
        }
        List<Throwable> failures = new ArrayList<Throwable>();
        List<Future<Void>> tasks = new LinkedList<Future<Void>>();
        for (PacoteImportacaoFile arquivo : pacoteIntegracaoBean.getArquivos()) {
            this.triggerFileProcess(tasks, failures, arquivo);
        }
        this.waitForOtherProcessors(tasks, failures);
        for (Throwable throwable : failures) {
            LOG.error("erro importando bean: " + pacoteIntegracaoBean, throwable);
        }
        if (!failures.isEmpty()) {
            throw new DataDigesterException(DataDigesterMessages.GENERIC_PROCESSOR_ERROR.create(), failures.get(0));
        }
    }

    /**
     * Dispara o processamento de um dos arquivos do pacote.
     * 
     * @param tasks tasks
     * @param failures failures
     * @param arquivo arquivo
     */
    private void triggerFileProcess(List<Future<Void>> tasks, List<Throwable> failures, PacoteImportacaoFile arquivo) {
        ProcessorAgent agent = new ProcessorAgent(arquivo);
        if (arquivo.isAsync()) {
            tasks.add(this.executionLogController.submitTransacted(agent));
        } else {
            this.waitForOtherProcessors(tasks, failures);
            tasks.add(this.executionLogController.submitTransacted(agent));
            this.waitForOtherProcessors(tasks, failures);
        }
    }

    /**
     * Espera que os outros processora executados assincronamente tenha terminado.
     * 
     * @param tasks as threads com os outros processors.
     * @param failures failures
     */
    private void waitForOtherProcessors(List<Future<Void>> tasks, List<Throwable> failures) {
        for (Future<Void> task : tasks) {
            try {
                task.get();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                failures.add(e);
            } catch (ExecutionException e) {
                failures.add(e.getCause());
            }
        }
        tasks.clear();
    }

    /**
     * Classe responsável por processar um arquivo em uma transação separada.
     * 
     * @author cesar
     */
    private class ProcessorAgent implements Callable<Void> {

        private final PacoteImportacaoFile arquivo;

        private final String fileName;

        private final String fileCode;

        /**
         * Construtor
         * 
         * @param arquivo arquivo
         */
        public ProcessorAgent(PacoteImportacaoFile arquivo) {
            this.arquivo = arquivo;
            this.fileName = this.arquivo.getFilename();
            this.fileCode = this.arquivo.getCodigo();
        }

        /**
         * Processa o arquivo.
         * 
         * @return o erro que ocorrer durante o processamento, ou <code>null</code> caso nenhum erro tenha ocorrido.
         * @throws Exception Exception
         */
        public Void call() throws Exception {
            InputStream data = null;
            try {
                data = this.arquivo.getData();
                PacoteImportacaoProcessor.LOG.debug("Processando arquivo '" + this.fileName + "', codigo: '" + this.fileCode + "'");
                PacoteImportacaoProcessor.this.dataDigesterController.process(this.fileCode, this.fileName, data);
            } catch (Exception e) {
                this.onError(e);
                throw e;
            } finally {
                IOUtils.closeQuietly(data);
                this.onSuccess();
            }

            return null;
        }

        /**
         * Registra o sucesso da execução
         */
        private void onSuccess() {
            try {
                if (PacoteImportacaoProcessor.this.executionLogController.isExecuting()) {
                    PacoteImportacaoProcessor.this.executionLogController.addSuccessfulItem("", "Processado arquivo: " + this.fileCode
                            + ":" + this.fileName);
                }
            } catch (Throwable t) {
                LOG.error(t);
            }
        }

        /**
         * Registra que a execução terminou com falha.
         * 
         * @param e exception
         */
        private void onError(Exception e) {
            try {
                PacoteImportacaoProcessor.LOG.error("Erro durante o processamento do arquivo '" + this.fileName + "', codigo: '"
                        + this.fileCode + "'", e);
                PacoteImportacaoProcessor.this.rtmController.monitor(e);
                if (PacoteImportacaoProcessor.this.executionLogController.isExecuting()) {
                    PacoteImportacaoProcessor.this.executionLogController.addErrorItem("", "Processado arquivo: " + this.fileCode + ":"
                            + this.fileName, e);
                }
            } catch (Throwable t2) {
                LOG.error(t2);
            }
        }

    }

    /**
     * @param dataDigesterController the dataDigesterController to set
     */
    public void setDataDigesterController(DataDigesterController dataDigesterController) {
        this.dataDigesterController = dataDigesterController;
    }

    /**
     * @param executionLogController the executionLogController to set
     */
    @Override
    public void setExecutionLogController(ExecutionLogController executionLogController) {
        this.executionLogController = executionLogController;
    }

    /**
     * @param rtmController the rtmController to set
     */
    public void setRtmController(RtmController rtmController) {
        this.rtmController = rtmController;
    }
}
